package com.swak.frame.ratelimit.interceptor;

import java.lang.reflect.Method;
import java.util.Collection;

import org.aopalliance.intercept.MethodInvocation;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.NoSuchBeanDefinitionException;
import org.springframework.beans.factory.NoUniqueBeanDefinitionException;
import org.springframework.lang.Nullable;

import com.swak.frame.command.SwakMethodProvider;
import com.swak.frame.command.fallback.SwakFallbackMethod;
import com.swak.frame.dto.base.SysRestCode;
import com.swak.frame.exception.SwakAssert;
import com.swak.frame.exception.ThrowableWrapper;
import com.swak.frame.interceptor.SwakAdviceSupport;
import com.swak.frame.interceptor.SwakOperationInvoker;
import com.swak.frame.ratelimit.adapter.LimitRaterAdapter;
import com.swak.frame.ratelimit.annotation.RateLimit;
import com.swak.frame.ratelimit.config.LimitOperation;
import com.swak.frame.ratelimit.exception.RateLimitException;

/**
 * ** AOP Alliance MethodInterceptor for declarative RateLimit management using the common RateLimit infrastructure
 * ({@link com.swak.frame.ratelimit.annotation.RateLimit}).
 * 
 * @ClassName: RateLimitInterceptor.java
 * @author: colley.ma
 * @date: 2022/01/26
 */
public class RateLimitInterceptor extends SwakAdviceSupport {

	private LimitRaterAdapter limitRaterAdapter;

	@Override
	public Object invoke(final MethodInvocation invocation) throws Throwable {
		Method method = invocation.getMethod();
		SwakOperationInvoker aopAllianceInvoker = () -> {
			try {
				return invocation.proceed();
			} catch (Throwable ex) {
				throw ThrowableWrapper.throwableWrapper(ex);
			}
		};
		Object target = invocation.getThis();
		SwakAssert.state(target != null, "Target must not be null");
		try {
			return execute(aopAllianceInvoker, target, method, invocation.getArguments());
		} catch (ThrowableWrapper th) {
			throw th.getOriginal();
		}
	}

	@Nullable
	protected Object execute(SwakOperationInvoker invoker, Object target, Method method, Object[] args) {
		// Check whether aspect is enabled (to cope with cases where the AJ is pulled in automatically)
		if (this.initialized) {
			Class<?> targetClass = getTargetClass(target);
			RaterOperationSource raterOperationSource = (RaterOperationSource) getOperationSource();
			if (raterOperationSource != null) {
				Collection<LimitOperation> operations = raterOperationSource.getBasicOperations(method, targetClass);
				if (!CollectionUtils.isEmpty(operations)) {
					LimitOperation limitOperation = operations.iterator().next();
					boolean isAcquire = limitRaterAdapter.tryAcquire(limitOperation);
					if (!isAcquire) {
						// 触发限流走fallback兜底或则返回限流码
						return getFallback(target,method,args,limitOperation.getRateLimit());
					}
					return invoker.invoke();
				}
			}
		}
		return super.invokeOperation(invoker);
	}

	protected Object getFallback(Object target, Method method, Object[] args, RateLimit rateLimit) {
		if (StringUtils.isEmpty(rateLimit.fallbackMethod())) {
			throw throwRateLimitException(rateLimit);
		}
		Class<?> targetClass = getTargetClass(target);
		SwakFallbackMethod fallbackMethod = SwakMethodProvider.getInstance().getFallbackMethod(targetClass, method,
				false);
		if (fallbackMethod.getMethod() == null) {
			logger.error("[swak-rateLimit] - No fallback available,fallbackMethod is null,fallbackMethodName:{}",
					fallbackMethod.getFallbackMethod());
			throw throwRateLimitException(rateLimit);
		}
		fallbackMethod.validateReturnType(method);
		try {
			return fallbackMethod.executeWithArgs(target, args);
		} catch (Throwable e) {
			logger.error("[swak-rateLimit] - getFallback execute error,fallbackMethodName:"
					+ fallbackMethod.getFallbackMethod(), e);
			throw throwRateLimitException(rateLimit);
		}
	}

	private RateLimitException throwRateLimitException(RateLimit rateLimit) {
		String retCode = StringUtils.firstNonEmpty(rateLimit.retCode(), SysRestCode.SYS_REQ_LIMIT.getRetCode());
		return new RateLimitException(retCode, SysRestCode.SYS_REQ_LIMIT.getRetMessage());
	}

	@Override
	public void afterSingletonsInstantiated() {
		super.afterSingletonsInstantiated();
		if (limitRaterAdapter == null) {
			// Lazily initialize limitRater Adapter...
			SwakAssert.state(this.beanFactory != null, "limitRaterAdapter or BeanFactory must be set on rateLimit aspect");
			try {
				setLimitRaterAdapter(this.beanFactory.getBean(LimitRaterAdapter.class));
			} catch (NoUniqueBeanDefinitionException ex) {
				throw new IllegalStateException("No LimitRaterAdapter specified, and no unique bean of type "
						+ "LimitRaterAdapter found. Mark one as primary or declare a specific LimitRaterAdapter to use.",
						ex);
			} catch (NoSuchBeanDefinitionException ex) {
				throw new IllegalStateException(
						"No LimitRaterAdapter specified, and no bean of type LimitRaterAdapter found. "
								+ "Register a LimitRaterAdapter bean or remove the @RateLimit annotation from your configuration.",
						ex);
			}
		}
		this.initialized = true;
	}

	public void setLimitRaterAdapter(LimitRaterAdapter limitRaterAdapter) {
		this.limitRaterAdapter = limitRaterAdapter;
	}
}
